package cn._51doit.kafka

import org.apache.kafka.clients.consumer.{ConsumerRecord, ConsumerRecords, KafkaConsumer}
import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}

import java.time.Duration
import java.util
import java.util.Properties

/**
 * @create: 2021-10-12 11:15
 * @author: 珠穆朗玛峰顶峰的男人
 * @program: kafka_java_scala
 * 消费者
 **/
object ScalaConsumer {
  def main(args: Array[String]): Unit = {
    //指定broker的地址
    val properties = new Properties
    // 连接kafka节点
    properties.setProperty("bootstrap.servers", "doit01:9092,doit02:9092,doit03:9092")
    //指定key反序列化方式
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    //指定value反序列化方式
    properties.setProperty("value.deserializer", classOf[StringDeserializer].getName) // 两种写法都行
    //指定消费者的组ID
    properties.setProperty("group.id", "r001")
    //指定读取数据的偏移量
    //earliest：如果没有历史偏移量，从最开始读取数据，如果对应的这个组ID记录过偏移量，会接着偏移量继续读
    //latest: 消费者启动后产生的数据才读取（默认的）
    //none: 如果指定group.id没有历史偏移量，抛出异常
    properties.setProperty("auto.offset.reset", "earliest")

    //默认情况下，消费者消息完数据后，会自动更新偏移量
    //消费者
    val kafkaConsumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](properties)

    //读取数据
    val topics = util.Arrays.asList("word")
    //订阅指定的topic，可以指定一个或多个topic
    kafkaConsumer.subscribe(topics)

    while (true){
      val kafkacon: ConsumerRecords[String, String] = kafkaConsumer.poll(Duration.ofSeconds(5))

      val it: util.Iterator[ConsumerRecord[String, String]] = kafkacon.iterator()

      while (it.hasNext) {
        val consumerrecord: ConsumerRecord[String, String] = it.next()

        println(consumerrecord)
      }
    }




  }
}
